package com.gzyj.flink.driver;

import com.alibaba.fastjson.JSON;
import com.gzyj.flink.driver.function.VehInfoProcessFun;
import com.gzyj.flink.driver.sink.DriverInfoSink;
import com.gzyj.flink.driver.source.mysql.VehicleSource;
import com.gzyj.flink.veh.AnbiaoVehicle;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Properties;

/**
 * @author lhj
 */
public class DriverApp {

    public static void main(String[] args) throws Exception {

        Configuration configuration = new Configuration();

        configuration.setString("mysql-url", "jdbc:mysql://127.0.0.1:3306/blade?useServerPrepStmts=false&rewriteBatchedStatements=true&useUnicode=true&serverTimezone=GMT%2B8&characterEncoding=utf-8&autoReconnect=true&failOverReadOnly=false&useCompression=true");
        configuration.setString("mysql-user", "root");
        configuration.setString("mysql-password", "123456");

        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        executionEnvironment.setParallelism(1);

        executionEnvironment.getConfig().setGlobalJobParameters(configuration);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");
        properties.setProperty("group.id", "flink");
        properties.setProperty("key.deserializer", StringDeserializer.class.getName());
        properties.setProperty("value.deserializer", StringDeserializer.class.getName());

        FlinkKafkaConsumerBase<String> kafkasource = new FlinkKafkaConsumer<String>("flink-driver", new SimpleStringSchema(), properties);

        DataStream<String> source = executionEnvironment.addSource(kafkasource);


        MapStateDescriptor<String, AnbiaoVehicle> vehicleMapState = new MapStateDescriptor<>("VehicleMapState", String.class, AnbiaoVehicle.class);

        BroadcastStream<AnbiaoVehicle> vehicleBroadcastStream = executionEnvironment.addSource(new VehicleSource())
                .broadcast(vehicleMapState);


        source.connect(vehicleBroadcastStream)
                .process(new VehInfoProcessFun(vehicleMapState))
                .addSink(new DriverInfoSink());

        executionEnvironment.execute("gzyj-daoluyunshu-driver");


    }
}
